package cn.itcast.mapreduce.db.write;

import cn.itcast.mapreduce.db.bean.GoodsBean;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * 读取文本数据写入到mysql中，在map阶段做读取文件数据 解析 封装成对象
 */
public class WriteDBMapper extends Mapper<LongWritable, Text, NullWritable, GoodsBean> {

    NullWritable outKet = NullWritable.get();
    GoodsBean outValue = new GoodsBean();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //获取两个计数器 用于统计合法数据 非法数据
        Counter sc = context.getCounter("mr_to_mysql", "Success");
        Counter fc = context.getCounter("mr_to_mysql", "Failed");

        String[] fields = value.toString().split("\\s+");
        //判断输入的数据字段是否有缺少 如果不满足需求 就是非法数据
        if (fields.length > 6) {
            //合法数据 提取字段封装成对象的属性
            outValue.set(
                    Long.parseLong(fields[1]),
                    fields[2],
                    fields[3],
                    Double.parseDouble(fields[4]),
                    Double.parseDouble(fields[5]),
                    Long.parseLong(fields[6])
            );
            context.write(outKet, outValue);

            //合法数据 计数器+1
            sc.increment(1);
        } else {
            //非法数据 计数器+1
            fc.increment(1);
        }

    }
}
